package skewreduce.framework;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;

import skewreduce.framework.physical.PhysicalOp;
import skewreduce.lib.Partition;
import skewreduce.lib.Partition.ID;


public class Schedule implements Comparable<Schedule>, Writable {
	final PhysicalOp.ID partitionID;
	final double expRuntime;
	final double cost;

	public Schedule(String line) {
		String[] flds = line.split("[,]");
		partitionID = new PhysicalOp.ID(flds[0]);
		expRuntime = Double.parseDouble(flds[1]);
		cost = Double.parseDouble(flds[2]);
	}
	
	public Schedule(PhysicalOp.ID id, double t, double c) {
		partitionID = id;
		this.expRuntime = t;
		this.cost = c;
	}
	
	public PhysicalOp.ID getID() { return partitionID; }
	public double getExpectedRuntime() { return expRuntime; }
	public double getCost() { return cost; }

	@Override
	public int hashCode() { return partitionID.hashCode(); }

	@Override
	public int compareTo(Schedule o) {
		if ( partitionID.getType() == o.partitionID.getType() ) {
			return (int)Math.signum(o.expRuntime - expRuntime);
		} else {
			return partitionID.compareTo(o.partitionID);
		}
	}
	
	@Override
	public String toString() {
		return String.format("%s,%e,%e",partitionID,expRuntime,cost);
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		throw new UnsupportedOperationException("Use createSchedule() instead");
	}

	@Override
	public void write(DataOutput out) throws IOException {
		partitionID.write(out);
		out.writeDouble(expRuntime);
		out.writeDouble(cost);
	}
	
	public static Schedule createSchedule(DataInput in) throws IOException {
		PhysicalOp.ID id = PhysicalOp.ID.createID(in);
		double t = in.readDouble();
		double c = in.readDouble();
		return new Schedule(id,t,c);
	}
}
